package com.atguigu.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/3/5
 *
 *  group window已经过时，强烈推荐使用 TVF window。
 *      TVF 对比 group window的优势:
 *          ①进行性能优化
 *          ②可以在窗口上应用topN
 *          ③支持标准版增强聚合 grouping sets
 *                  grouping sets(): hive，flink， ck不支持
 *                  cube : hive，flink,ck支持
 *                  rollup: hive,flink,ck支持
 *                  withTotal: 只有ck支持，hive不支持，flink不支持
 *          ④增加 累积聚合 窗口
 *
 *          语法区别： 在Table 关键字定义窗口
 *
 *    TVF window不支持 session窗口！
 *
 */
public class Demo11_TVFWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = " create table t1 ( id string, ts bigint  , vc int , " +
            "                   et  as TO_TIMESTAMP_LTZ(ts,3)    ,   " +
            "                   pt  as  PROCTIME() ," +
            "                   WATERMARK FOR et AS et - INTERVAL '1' SECOND   )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        tableEnvironment.executeSql(createTableSql);

        /*
                TABLE( 窗口函数  )
                   定义窗口的函数:
                        TUMBLE(TABLE 表名, DESCRIPTOR(时间属性), SIZE )
                        HOP(TABLE 表名, DESCRIPTOR(时间属性), SLIDE ,SIZE )
         */
        String sql1 = " select  window_start, window_end, id,sum(vc) sumVc " +
            "  from TABLE(" +
            "    TUMBLE(TABLE t1, DESCRIPTOR(et), INTERVAL '5' SECONDS)) " +
            "  GROUP BY window_start, window_end,id ";


        // HOP table function based aggregate requires size must be an integral multiple of slide
        // TVF 要求 size 是 slide的整倍数
        String sql2 = " select  window_start, window_end, id,sum(vc) sumVc " +
            "  from TABLE( HOP(TABLE t1, DESCRIPTOR(et), INTERVAL '3' SECONDS , INTERVAL '6' SECONDS) ) " +
            "  GROUP BY window_start, window_end,id ";

        tableEnvironment.sqlQuery(sql2)
                        .execute()
                        .print();

    }
}
